package com.jenrey.hbase.dao.impl;

import com.jenrey.bean.CategoryClickCount;
import com.jenrey.hbase.dao.HBaseDao;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * HBaseAPI的实现类
 */
public class HBaseImpl implements HBaseDao {
    //获取数据库连接,类似于JDBC里面的连接池的概念
    HConnection htablePool = null;

    //只要一new就执行下面构造方法里面的代码，即创建并连接HBase成功
    public HBaseImpl() {
        //创建HBase的配置文件
        Configuration conf = HBaseConfiguration.create();
        //设置zookeeper
        conf.set("hbase.zookeeper.quorum", "hadoop02:2181");
//        conf.set("zookeeper.znode.parent","/myhbase");
        try {
            //创建连接
            htablePool = HConnectionManager.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    /**
     * TODO:根据表名获取表对象
     *
     * @param tableName 表名
     * @return 表对象
     * HTableInterface  得到这个表的名称（HBase自带API）
     */
    public HTableInterface getTable(String tableName) {
        HTableInterface table = null;
        try {
            //根据表名获取表对象
            table = htablePool.getTable(tableName);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return table;
    }

    //

    /**
     * TODO:实现往hbase里面插入一条数据
     *
     * @param tableName 表名
     * @param rowkey    rowkey
     * @param family    列簇
     * @param q         品类
     * @param value     出现了多少次
     *                  hbase:只有一种数据类型，字节数组
     */
    @Override
    public void save(String tableName, String rowkey, String family, String q, long value) {
        HTableInterface table = getTable(tableName);
        try {
            //incrementColumnValue会自动帮我们累加value的值。不需要在读取、累加、返回值
            table.incrementColumnValue(rowkey.getBytes(), family.getBytes(), q.getBytes(), value);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (table != null) {
                try {
                    table.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    //

    /**
     * TODO:实现按照表名和rowkey查询数据
     *
     * @param tableName 表名
     * @param rowkey    rowkey
     * @return
     */
    @Override
    public List<CategoryClickCount> count(String tableName, String rowkey) {
        //我们要的是电影和电视剧出现的次数，所以要封装成一个对象存到List集合中
        ArrayList<CategoryClickCount> list = new ArrayList<>();
        //获取表对象
        HTableInterface table = getTable(tableName);
        //前缀过滤器
        PrefixFilter prefixFilter = new PrefixFilter(rowkey.getBytes());
        //创建扫描仪
        Scan scan = new Scan();
        //给扫描仪安装过滤器
        scan.setFilter(prefixFilter);
        //拿着扫描仪去获取数据
        try {
            ResultScanner scanner = table.getScanner(scan);
            //遍历集合，一个结果就是一个Result对象
            for (Result result : scanner) {
                //我们要的是电影和电视剧出现的次数，所以要封装成一个对象存到List集合中,既是CategoryClickCount对象，name value
                //rowkey date_name

                //遍历里面的每一个东西，rawCells把里面的每一个东西都看成一个单元格
                for(Cell cell:result.rawCells()){
                    byte[] date_name = CellUtil.cloneRow(cell);
                    String name = new String(date_name).split("_")[1];
                    byte[] value = CellUtil.cloneValue(cell);
                    long count = Bytes.toLong(value);
                    CategoryClickCount categoryClickCount = new CategoryClickCount(name, count);
                    list.add(categoryClickCount);
                }

            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (table != null) {
                try {
                    table.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        }
        return list;
    }
}
